// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: support for request discovery
// feature: provide query interfaces
// feature: Split history as an independent common module
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: discovery support client and server
// feature: add exception code
// ------------------------------------------------------------------

#include <discovery/endpoint/EDPSimple.h>

#include <discovery/endpoint/EDPSimpleListeners.h>
#include <discovery/participant/PDP.h>
#include <rtps/participant/RTPSParticipantImpl.h>
#include <edds/rtps/writer/StatefulWriter.h>
#include <edds/rtps/reader/StatefulReader.h>
#include <edds/rtps/attributes/HistoryAttributes.h>
#include <edds/rtps/attributes/WriterAttributes.h>
#include <edds/rtps/attributes/ReaderAttributes.h>
#include <history/ReaderHistory.h>
#include <history/WriterHistory.h>
#include <edds/rtps/builtin/data/WriterProxyData.h>
#include <edds/rtps/builtin/data/ReaderProxyData.h>
#include <edds/rtps/builtin/data/ParticipantProxyData.h>
#include <edds/rtps/builtin/BuiltinProtocols.h>
#include <edds/rtps/builtin/data/BuiltinEndpoints.hpp>

#include <mutex>
#include <forward_list>
#include <algorithm>

#include <edds/dds/core/policy/ParameterSerializer.hpp>

#include <elog/Log.hpp>

#include <history/TopicPayloadPoolRegistry.hpp>

#include <discovery/endpoint/EDPUtils.hpp>

using ParameterList = evbs::edds::dds::ParameterList;
using namespace vbs::common;
namespace evbs {
namespace ertps {
namespace rtps {

// Default configuration values for EDP entities.
static const Duration_t edp_heartbeat_period {0, 800U * 1000U * 1000U};   // 800ms
static const Duration_t edp_nack_response_delay {0, 1U * 1000U * 1000U};  // 1ms
static const Duration_t edp_nack_supression_duration {0, 100U * 1000U};   // 100us
static const Duration_t edp_heartbeat_response_delay {0, 100U * 1000U};   // 100us

static const int32_t edp_reader_initial_reserved_caches = 1;
static const int32_t edp_writer_initial_reserved_caches = 20;

static void delete_reader(RTPSParticipantImpl* participant, std::pair<StatefulReader*, ReaderHistory*>& reader_pair,
                          std::shared_ptr<vbs::common::ITopicPayloadPool>& pool) {
    if (nullptr != reader_pair.first) {
        (void)participant->deleteUserEndpoint(reader_pair.first->getGuid());
        EDPUtils::release_payload_pool(pool, reader_pair.second->m_att, true);
        delete (reader_pair.second);
    }
}

static void delete_writer(RTPSParticipantImpl* participant, std::pair<StatefulWriter*, WriterHistory*>& writer_pair,
                          std::shared_ptr<vbs::common::ITopicPayloadPool>& pool) {
    if (nullptr != writer_pair.first) {
        (void)participant->deleteUserEndpoint(writer_pair.first->getGuid());
        EDPUtils::release_payload_pool(pool, writer_pair.second->m_att, false);
        delete (writer_pair.second);
    }
}

EDPSimple::EDPSimple(PDP* p, RTPSParticipantImpl* part)
    : EDP(p, part), publications_listener_(nullptr), subscriptions_listener_(nullptr) {}

EDPSimple::~EDPSimple() {
#if HAVE_SECURITY
    delete_reader(mp_RTPSParticipant, publications_secure_reader_, sec_pub_reader_payload_pool_);
    delete_reader(mp_RTPSParticipant, subscriptions_secure_reader_, sec_sub_reader_payload_pool_);

    delete_writer(mp_RTPSParticipant, publications_secure_writer_, sec_pub_writer_payload_pool_);
    delete_writer(mp_RTPSParticipant, subscriptions_secure_writer_, sec_sub_writer_payload_pool_);
#endif  // if HAVE_SECURITY

    delete_reader(mp_RTPSParticipant, publications_reader_, pub_reader_payload_pool_);
    delete_reader(mp_RTPSParticipant, subscriptions_reader_, sub_reader_payload_pool_);

    delete_writer(mp_RTPSParticipant, publications_writer_, pub_writer_payload_pool_);
    delete_writer(mp_RTPSParticipant, subscriptions_writer_, sub_writer_payload_pool_);

    if (nullptr != publications_listener_) {
        delete (publications_listener_);
    }

    if (nullptr != subscriptions_listener_) {
        delete (subscriptions_listener_);
    }
}

bool EDPSimple::initEDP(BuiltinAttributes& attributes) {
    logDebug(RTPS_EDP, "Beginning Simple Endpoint Discovery Protocol");
    m_discovery = attributes;

    if (!createSEDPEndpoints()) {
        elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR, "Problem creation SimpleEDP endpoints");
        return false;
    }

#if HAVE_SECURITY
    if (mp_RTPSParticipant->is_secure() && (!create_sedp_secure_endpoints())) {
        elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR, "Problem creation SimpleEDP endpoints");
        return false;
    }
#endif  // if HAVE_SECURITY

    return true;
}

EDPSimple::t_p_StatefulWriter EDPSimple::get_builtin_writer_history_pair_by_entity(const EntityId_t& entity_id) {
    t_p_StatefulWriter ret {};

    if (entity_id == c_EntityId_SEDPPubWriter) {
        ret = publications_writer_;

    } else if (entity_id == c_EntityId_SEDPSubWriter) {
        ret = subscriptions_writer_;
    }
#if HAVE_SECURITY
    else if (entity_id == sedp_builtin_publications_secure_writer) {
        ret = publications_secure_writer_;
    } else if (entity_id == sedp_builtin_subscriptions_secure_writer) {
        ret = subscriptions_secure_writer_;
    }
#endif  // HAVE_SECURITY
    else {
        elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR, "Could not find the requested writer builtin endpoint");
    }

    return ret;
}

EDPSimple::t_p_StatefulReader EDPSimple::get_builtin_reader_history_pair_by_entity(const EntityId_t& entity_id) {
    t_p_StatefulReader ret {nullptr, nullptr};

    if (entity_id == c_EntityId_SEDPPubReader || entity_id == c_EntityId_SEDPPubWriter) {
        ret = publications_reader_;
    } else if (entity_id == c_EntityId_SEDPSubReader || entity_id == c_EntityId_SEDPSubWriter) {
        ret = subscriptions_reader_;
    }
#if HAVE_SECURITY
    else if (entity_id == sedp_builtin_publications_secure_reader ||
             entity_id == sedp_builtin_publications_secure_writer) {
        ret = publications_secure_reader_;
    } else if (entity_id == sedp_builtin_subscriptions_secure_reader ||
               entity_id == sedp_builtin_subscriptions_secure_writer) {
        ret = subscriptions_secure_reader_;
    }
#endif  // HAVE_SECURITY
    else {
        elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR, "Could not find the requested reader builtin endpoint");
    }

    return ret;
}

void EDPSimple::set_builtin_reader_history_attributes(HistoryAttributes& attributes) {
    attributes.initialReservedCaches = edp_reader_initial_reserved_caches;
    attributes.payloadMaxSize = mp_PDP->builtin_attributes().readerPayloadSize;
    attributes.memoryPolicy = mp_PDP->builtin_attributes().readerHistoryMemoryPolicy;
}

void EDPSimple::set_builtin_writer_history_attributes(HistoryAttributes& attributes) {
    attributes.initialReservedCaches = edp_writer_initial_reserved_caches;
    attributes.payloadMaxSize = mp_PDP->builtin_attributes().writerPayloadSize;
    attributes.memoryPolicy = mp_PDP->builtin_attributes().writerHistoryMemoryPolicy;
}

void EDPSimple::set_builtin_reader_attributes(ReaderAttributes& attributes) {
    const RTPSParticipantAttributes& pattr = mp_PDP->getRTPSParticipant()->getRTPSParticipantAttributes();

    // Matched writers will depend on total number of participants
    attributes.matched_writers_allocation = pattr.allocation.participants;

    // As participants allocation policy includes the local participant, one has to be substracted
    if (attributes.matched_writers_allocation.initial > 1U) {
        attributes.matched_writers_allocation.initial--;
    }
    if ((attributes.matched_writers_allocation.maximum > 1U) &&
        (attributes.matched_writers_allocation.maximum < std::numeric_limits<size_t>::max())) {
        attributes.matched_writers_allocation.maximum--;
    }

    // Locators are copied from the local participant metatraffic locators
    attributes.endpoint.unicastLocatorList.clear();
    for (const Locator_t& loc : mp_PDP->getLocalParticipantProxyData()->metatraffic_locators.unicast) {
        attributes.endpoint.unicastLocatorList.push_back(loc);
    }
    attributes.endpoint.multicastLocatorList.clear();
    for (const Locator_t& loc : mp_PDP->getLocalParticipantProxyData()->metatraffic_locators.multicast) {
        attributes.endpoint.multicastLocatorList.push_back(loc);
    }
    attributes.endpoint.external_unicast_locators = mp_PDP->builtin_attributes().metatraffic_external_unicast_locators;
    attributes.endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;

    // Timings are configured using EDP default values
    attributes.times.heartbeatResponseDelay = edp_heartbeat_response_delay;

    // EDP endpoints are always reliable, transsient local, keyed topics
    attributes.endpoint.reliabilityKind = RELIABLE;
    attributes.endpoint.durabilityKind = TRANSIENT_LOCAL;
    attributes.endpoint.topicKind = WITH_KEY;
}

void EDPSimple::set_builtin_writer_attributes(WriterAttributes& attributes) {
    const RTPSParticipantAttributes& pattr = mp_PDP->getRTPSParticipant()->getRTPSParticipantAttributes();

    // Matched readers will depend on total number of participants
    attributes.matched_readers_allocation = pattr.allocation.participants;

    // As participants allocation policy includes the local participant, one has to be substracted
    if (attributes.matched_readers_allocation.initial > 1U) {
        attributes.matched_readers_allocation.initial--;
    }
    if ((attributes.matched_readers_allocation.maximum > 1U) &&
        (attributes.matched_readers_allocation.maximum < std::numeric_limits<size_t>::max())) {
        attributes.matched_readers_allocation.maximum--;
    }

    // Locators are copied from the local participant metatraffic locators
    attributes.endpoint.unicastLocatorList.clear();
    for (const Locator_t& loc : mp_PDP->getLocalParticipantProxyData()->metatraffic_locators.unicast) {
        attributes.endpoint.unicastLocatorList.push_back(loc);
    }
    attributes.endpoint.multicastLocatorList.clear();
    for (const Locator_t& loc : mp_PDP->getLocalParticipantProxyData()->metatraffic_locators.multicast) {
        attributes.endpoint.multicastLocatorList.push_back(loc);
    }
    attributes.endpoint.external_unicast_locators = mp_PDP->builtin_attributes().metatraffic_external_unicast_locators;
    attributes.endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;

    // Timings are configured using EDP default values
    attributes.times.heartbeatPeriod = edp_heartbeat_period;
    attributes.times.nackResponseDelay = edp_nack_response_delay;
    attributes.times.nackSupressionDuration = edp_nack_supression_duration;

    // EDP endpoints are always reliable, transsient local, keyed topics
    attributes.endpoint.reliabilityKind = RELIABLE;
    attributes.endpoint.durabilityKind = TRANSIENT_LOCAL;
    attributes.endpoint.topicKind = WITH_KEY;
}

bool EDPSimple::createSEDPEndpoints() {
    WriterAttributes watt;
    ReaderAttributes ratt;
    HistoryAttributes reader_history_att;
    HistoryAttributes writer_history_att;

    set_builtin_reader_history_attributes(reader_history_att);
    set_builtin_writer_history_attributes(writer_history_att);
    set_builtin_reader_attributes(ratt);
    set_builtin_writer_attributes(watt);

    publications_listener_ = new EDPSimplePUBListener(this);
    subscriptions_listener_ = new EDPSimpleSUBListener(this);

    if (m_discovery.discovery_config.m_simpleEDP.use_PublicationWriterANDSubscriptionReader) {
        if (!EDPUtils::create_edp_writer(mp_RTPSParticipant, "DCPSPublications", c_EntityId_SEDPPubWriter,
                                         writer_history_att, watt, publications_listener_, pub_writer_payload_pool_,
                                         publications_writer_)) {
            return false;
        }

        logDebug(RTPS_EDP, mp_RTPSParticipant->getGuid() << " SEDP Publication Writer created");

        if (!EDPUtils::create_edp_reader(mp_RTPSParticipant, "DCPSSubscriptions", c_EntityId_SEDPSubReader,
                                         reader_history_att, ratt, subscriptions_listener_, sub_reader_payload_pool_,
                                         subscriptions_reader_)) {
            return false;
        }

        logDebug(RTPS_EDP, mp_RTPSParticipant->getGuid() << " SEDP Subscription Reader created");
    }

    if (m_discovery.discovery_config.m_simpleEDP.use_PublicationReaderANDSubscriptionWriter) {
        if (!EDPUtils::create_edp_reader(mp_RTPSParticipant, "DCPSPublications", c_EntityId_SEDPPubReader,
                                         reader_history_att, ratt, publications_listener_, pub_reader_payload_pool_,
                                         publications_reader_)) {
            return false;
        }

        logDebug(RTPS_EDP, mp_RTPSParticipant->getGuid() << " SEDP Publication Reader created");

        if (!EDPUtils::create_edp_writer(mp_RTPSParticipant, "DCPSSubscriptions", c_EntityId_SEDPSubWriter,
                                         writer_history_att, watt, subscriptions_listener_, sub_writer_payload_pool_,
                                         subscriptions_writer_)) {
            return false;
        }

        logDebug(RTPS_EDP, mp_RTPSParticipant->getGuid() << " SEDP Subscription Writer created");
    }

    logDebug(RTPS_EDP, "SEDP Endpoints creation finished");
    return true;
}

#if HAVE_SECURITY
bool EDPSimple::create_sedp_secure_endpoints() {
    WriterAttributes watt;
    ReaderAttributes ratt;
    HistoryAttributes reader_history_att;
    HistoryAttributes writer_history_att;

    set_builtin_reader_history_attributes(reader_history_att);
    set_builtin_writer_history_attributes(writer_history_att);
    set_builtin_reader_attributes(ratt);
    set_builtin_writer_attributes(watt);
    mp_PDP->add_builtin_security_attributes(ratt, watt);

    if (m_discovery.discovery_config.m_simpleEDP.enable_builtin_secure_publications_writer_and_subscriptions_reader) {
        if (!EDPUtils::create_edp_writer(mp_RTPSParticipant, "DCPSPublicationsSecure",
                                         sedp_builtin_publications_secure_writer, writer_history_att, watt,
                                         publications_listener_, sec_pub_writer_payload_pool_,
                                         publications_secure_writer_)) {
            return false;
        }

        logDebug(RTPS_EDP, mp_RTPSParticipant->getGuid() << " SEDP Publication Writer created");

        if (!EDPUtils::create_edp_reader(mp_RTPSParticipant, "DCPSSubscriptionsSecure",
                                         sedp_builtin_subscriptions_secure_reader, reader_history_att, ratt,
                                         subscriptions_listener_, sec_sub_reader_payload_pool_,
                                         subscriptions_secure_reader_)) {
            return false;
        }

        logDebug(RTPS_EDP, mp_RTPSParticipant->getGuid() << " SEDP Subscription Reader created");
    }

    if (m_discovery.discovery_config.m_simpleEDP.enable_builtin_secure_subscriptions_writer_and_publications_reader) {
        if (!EDPUtils::create_edp_reader(mp_RTPSParticipant, "DCPSPublicationsSecure",
                                         sedp_builtin_publications_secure_reader, reader_history_att, ratt,
                                         publications_listener_, sec_pub_reader_payload_pool_,
                                         publications_secure_reader_)) {
            return false;
        }

        logDebug(RTPS_EDP, mp_RTPSParticipant->getGuid() << " SEDP Publication Reader created");

        if (!EDPUtils::create_edp_writer(mp_RTPSParticipant, "DCPSSubscriptionsSecure",
                                         sedp_builtin_subscriptions_secure_writer, writer_history_att, watt,
                                         subscriptions_listener_, sec_sub_writer_payload_pool_,
                                         subscriptions_secure_writer_)) {
            return false;
        }

        logDebug(RTPS_EDP, mp_RTPSParticipant->getGuid() << " SEDP Subscription Writer created");
    }
    logDebug(RTPS_EDP, "SEDP Endpoints secure creation finished");
    return true;
}

#endif  // if HAVE_SECURITY

bool EDPSimple::processLocalReaderProxyData(RTPSReader* local_reader, ReaderProxyData* rdata) {
    auto* writer = &subscriptions_writer_;

#if HAVE_SECURITY
    if (local_reader->getAttributes().security_attributes().is_discovery_protected) {
        writer = &subscriptions_secure_writer_;
    }
#endif  // if HAVE_SECURITY
    CacheChange_t* change = nullptr;
    bool ret_val = serialize_reader_proxy_data(*rdata, *writer, true, &change);
    if (change != nullptr) {
        logDebug(RTPS_EDP, " add rdata into CacheChanges, guid: " << rdata->guid()
                                                                  << " topicname: " << rdata->typeName()
                                                                  << " topickind: " << rdata->topicKind());
        WriteParams wp;
        SampleIdentity local;
        local.writer_guid(writer->first->getGuid());
        local.sequence_number(writer->second->next_sequence_number());
        wp.sample_identity(local);
        wp.related_sample_identity(local);

        (void)writer->second->add_change(change, wp);
    }
    if (local_reader->getSendMulti()) {  //LCOV_EXCL_START
        auto* subwriter = &subscriptions_writer_;
        if (subwriter->first != nullptr) {
            subwriter->first->setSendMulti(true);
        }
        auto* pubwriter = &publications_writer_;
        if (pubwriter->first != nullptr) {
            pubwriter->first->setSendMulti(true);
        }
        auto* pubreader = &publications_reader_;
        if (pubreader->first != nullptr) {
            pubreader->first->setSendMulti(true);
        }
        auto* subreader = &subscriptions_reader_;
        if (subreader->first != nullptr) {
            subreader->first->setSendMulti(true);
        }
#if HAVE_SECURITY
        if (local_reader->getAttributes().security_attributes().is_discovery_protected) {
            auto* subwriter = &subscriptions_secure_writer_;
            if (subwriter->first != nullptr) {
                subwriter->first->setSendMulti(true);
            }
            auto* pubwriter = &publications_secure_writer_;
            if (pubwriter->first != nullptr) {
                pubwriter->first->setSendMulti(true);
            }
            auto* pubreader = &publications_secure_reader_;
            if (pubreader->first != nullptr) {
                pubreader->first->setSendMulti(true);
            }
            auto* subreader = &subscriptions_secure_reader_;
            if (subreader->first != nullptr) {
                subreader->first->setSendMulti(true);
            }
        }  //LCOV_EXCL_STOP
#endif     // if HAVE_SECURITY
    }
    return ret_val;
}

bool EDPSimple::processLocalWriterProxyData(RTPSWriter* local_writer, WriterProxyData* wdata) {
    auto* writer = &publications_writer_;

#if HAVE_SECURITY
    if (local_writer->getAttributes().security_attributes().is_discovery_protected) {
        writer = &publications_secure_writer_;
    }
#endif  // if HAVE_SECURITY

    CacheChange_t* change = nullptr;
    bool ret_val = serialize_writer_proxy_data(*wdata, *writer, true, &change);
    if (change != nullptr) {
        logDebug(RTPS_EDP, " add wdata into CacheChanges, guid: " << wdata->guid()
                                                                  << " topicname: " << wdata->typeName()
                                                                  << " topickind: " << wdata->topicKind());
        WriteParams wp;
        SampleIdentity local;
        local.writer_guid(writer->first->getGuid());
        local.sequence_number(writer->second->next_sequence_number());
        wp.sample_identity(local);
        wp.related_sample_identity(local);

        (void)writer->second->add_change(change, wp);
    }
    if (local_writer->getSendMulti()) {  //LCOV_EXCL_START
        auto* pubwriter = &publications_writer_;
        if (pubwriter->first != nullptr) {
            pubwriter->first->setSendMulti(true);
        }
        auto* subwriter = &subscriptions_writer_;
        if (subwriter->first != nullptr) {
            subwriter->first->setSendMulti(true);
        }
        auto* pubreader = &publications_reader_;
        if (pubreader->first != nullptr) {
            pubreader->first->setSendMulti(true);
        }
        auto* subreader = &subscriptions_reader_;
        if (subreader->first != nullptr) {
            subreader->first->setSendMulti(true);
        }
#if HAVE_SECURITY
        if (local_writer->getAttributes().security_attributes().is_discovery_protected) {
            auto* subwriter = &subscriptions_secure_writer_;
            if (subwriter->first != nullptr) {
                subwriter->first->setSendMulti(true);
            }
            auto* pubwriter = &publications_secure_writer_;
            if (pubwriter->first != nullptr) {
                pubwriter->first->setSendMulti(true);
            }
            auto* pubreader = &publications_secure_reader_;
            if (pubreader->first != nullptr) {
                pubreader->first->setSendMulti(true);
            }
            auto* subreader = &subscriptions_secure_reader_;
            if (subreader->first != nullptr) {
                subreader->first->setSendMulti(true);
            }
        }  //LCOV_EXCL_STOP
#endif     // if HAVE_SECURITY
    }
    return ret_val;
}

bool EDPSimple::serialize_writer_proxy_data(const WriterProxyData& data, const t_p_StatefulWriter& writer,
                                            bool remove_same_instance, CacheChange_t** created_change) {
    return serialize_proxy_data(data, writer, remove_same_instance, created_change);
}

bool EDPSimple::serialize_reader_proxy_data(const ReaderProxyData& data, const t_p_StatefulWriter& writer,
                                            bool remove_same_instance, CacheChange_t** created_change) {
    return serialize_proxy_data(data, writer, remove_same_instance, created_change);
}

template <typename ProxyData>
bool EDPSimple::serialize_proxy_data(const ProxyData& data, const t_p_StatefulWriter& writer, bool remove_same_instance,
                                     CacheChange_t** created_change) {
    assert(created_change != nullptr);
    *created_change = nullptr;

    if (writer.first != nullptr) {
        uint32_t cdr_size = data.get_serialized_size(true);
        CacheChange_t* change =
            writer.first->new_change([cdr_size]() -> uint32_t { return cdr_size; }, ALIVE, data.key());
        if (change != nullptr) {
            CDRMessage_t aux_msg(change->serializedPayload);

#if __BIG_ENDIAN__
            change->serializedPayload.encapsulation = (uint16_t)PL_CDR_BE;
            aux_msg.msg_endian = BIGEND;
#else
            change->serializedPayload.encapsulation = (uint16_t)PL_CDR_LE;
            aux_msg.msg_endian = LITTLEEND;
#endif  // if __BIG_ENDIAN__

            (void)data.writeToCDRMessage(&aux_msg, true);
            change->serializedPayload.length = (uint16_t)aux_msg.length;

            if (remove_same_instance) {
                std::unique_lock<RecursiveTimedMutex> lock(writer.second->getMutex());
                for (auto ch = writer.second->changesBegin(); ch != writer.second->changesEnd(); ++ch) {
                    if ((*ch)->instanceHandle == change->instanceHandle) {
                        (void)writer.second->remove_change_nts(*ch);
                        break;
                    }
                }
            }
            *created_change = change;
            return true;
        }
        return false;
    }
    return true;
}

bool EDPSimple::removeLocalWriter(RTPSWriter* W) {
    logDebug(RTPS_EDP, W->getGuid());

    auto* writer = &publications_writer_;

#if HAVE_SECURITY
    if (W->getAttributes().security_attributes().is_discovery_protected) {
        writer = &publications_secure_writer_;
    }
#endif  // if HAVE_SECURITY

    if (writer->first != nullptr) {
        InstanceHandle_t iH;
        iH = W->getGuid();
        CacheChange_t* change =
            writer->first->new_change([this]() -> uint32_t { return mp_PDP->builtin_attributes().writerPayloadSize; },
                                      NOT_ALIVE_DISPOSED_UNREGISTERED, iH);
        if (change != nullptr) {
            {
                std::unique_lock<RecursiveTimedMutex> history_lock(writer->second->getMutex());
                for (auto ch = writer->second->changesBegin(); ch != writer->second->changesEnd(); ++ch) {
                    if ((*ch)->instanceHandle == change->instanceHandle) {
                        writer->second->remove_change_nts(*ch);
                        break;
                    }
                }
            }

            (void)writer->second->add_change(change);
        }
    }
    return mp_PDP->removeWriterProxyData(W->getGuid());
}

bool EDPSimple::removeLocalReader(RTPSReader* R) {
    logDebug(RTPS_EDP, R->getGuid());

    auto* writer = &subscriptions_writer_;

#if HAVE_SECURITY
    if (R->getAttributes().security_attributes().is_discovery_protected) {
        writer = &subscriptions_secure_writer_;
    }
#endif  // if HAVE_SECURITY

    if (writer->first != nullptr) {
        InstanceHandle_t iH;
        iH = (R->getGuid());
        CacheChange_t* change =
            writer->first->new_change([this]() -> uint32_t { return mp_PDP->builtin_attributes().writerPayloadSize; },
                                      NOT_ALIVE_DISPOSED_UNREGISTERED, iH);
        if (change != nullptr) {
            {
                std::unique_lock<RecursiveTimedMutex> history_lock(writer->second->getMutex());
                for (auto ch = writer->second->changesBegin(); ch != writer->second->changesEnd(); ++ch) {
                    if ((*ch)->instanceHandle == change->instanceHandle) {
                        writer->second->remove_change_nts(*ch);
                        break;
                    }
                }
            }

            (void)writer->second->add_change(change);
        }
    }
    return mp_PDP->removeReaderProxyData(R->getGuid());
}

void EDPSimple::assignRemoteEndpoints(const ParticipantProxyData& pdata, bool assign_secure_endpoints,
                                      bool is_request_enable) {
    uint32_t endp = pdata.m_availableBuiltinEndpoints;
    uint32_t auxendp = endp;
    bool use_multicast_locators = !mp_PDP->getRTPSParticipant()->getAttributes().builtin.avoid_builtin_multicast;

    if (!use_multicast_locators) {
        if (pdata.metatraffic_locators.unicast.empty()) {
            logWarning(RTPS_EDP,
                       "No unicast locator to remote participant, don't assign remote SimpleEDP endpoints for "
                           << pdata.m_guid);
            return;
        }
    }

    auxendp &= DISC_BUILTIN_ENDPOINT_PUBLICATION_ANNOUNCER;

    auto temp_reader_proxy_data = get_temporary_reader_proxies_pool().get();

    temp_reader_proxy_data->clear();
    temp_reader_proxy_data->guid().guidPrefix = pdata.m_guid.guidPrefix;
    temp_reader_proxy_data->set_remote_locators(pdata.metatraffic_locators, mp_RTPSParticipant, use_multicast_locators);
    temp_reader_proxy_data->m_qos.m_durability.kind = TRANSIENT_LOCAL_DURABILITY_QOS;
    temp_reader_proxy_data->m_qos.m_reliability.kind = RELIABLE_RELIABILITY_QOS;
    if (is_request_enable) {
        elogInfo(RTPS_EDP, "It is compatible remote, enable request. remote reader GUID: " << pdata.m_guid);
        temp_reader_proxy_data->is_request_enable(is_request_enable);
    }

    auto temp_writer_proxy_data = get_temporary_writer_proxies_pool().get();

    temp_writer_proxy_data->clear();
    temp_writer_proxy_data->guid().guidPrefix = pdata.m_guid.guidPrefix;
    temp_writer_proxy_data->persistence_guid(pdata.get_persistence_guid());
    temp_writer_proxy_data->set_remote_locators(pdata.metatraffic_locators, mp_RTPSParticipant, use_multicast_locators);
    temp_writer_proxy_data->m_qos.m_durability.kind = TRANSIENT_LOCAL_DURABILITY_QOS;
    temp_writer_proxy_data->m_qos.m_reliability.kind = RELIABLE_RELIABILITY_QOS;

    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (publications_reader_.first != nullptr))  // Exist Pub Writer and i have pub reader
    {
        temp_writer_proxy_data->guid().entityId = c_EntityId_SEDPPubWriter;
        logDebug(RTPS_EDP, "Adding SEDP Pub Writer " << temp_writer_proxy_data->guid() << " to my Pub Reader "
                                                     << publications_reader_.first->getGuid());
        temp_writer_proxy_data->set_persistence_entity_id(c_EntityId_SEDPPubWriter);
        (void)publications_reader_.first->matched_writer_add(*temp_writer_proxy_data);
    }
    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_PUBLICATION_DETECTOR;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (publications_writer_.first != nullptr))  // Exist Pub Detector
    {
        temp_reader_proxy_data->guid().entityId = c_EntityId_SEDPPubReader;
        logDebug(RTPS_EDP, "Adding SEDP Pub Reader " << temp_reader_proxy_data->guid() << " to my Pub Writer "
                                                     << publications_writer_.first->getGuid());
        (void)publications_writer_.first->matched_reader_add(*temp_reader_proxy_data);
    }
    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_ANNOUNCER;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (subscriptions_reader_.first != nullptr))  // Exist Pub Announcer
    {
        temp_writer_proxy_data->guid().entityId = c_EntityId_SEDPSubWriter;
        logDebug(RTPS_EDP, "Adding SEDP Sub Writer " << temp_writer_proxy_data->guid() << " to my Sub Reader "
                                                     << subscriptions_reader_.first->getGuid());
        temp_writer_proxy_data->set_persistence_entity_id(c_EntityId_SEDPSubWriter);
        (void)subscriptions_reader_.first->matched_writer_add(*temp_writer_proxy_data);
    }
    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_DETECTOR;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (subscriptions_writer_.first != nullptr))  // Exist Pub Announcer
    {
        temp_reader_proxy_data->guid().entityId = c_EntityId_SEDPSubReader;
        logDebug(RTPS_EDP, "Adding SEDP Sub Reader " << temp_reader_proxy_data->guid() << " to my Sub Writer "
                                                     << subscriptions_writer_.first->getGuid());
        (void)subscriptions_writer_.first->matched_reader_add(*temp_reader_proxy_data);
    }

#if HAVE_SECURITY
    // m核不进行一下匹配
    //  if (!mp_RTPSParticipant->security_manager().isMvbs(pdata.m_VendorId)) {
    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_PUBLICATION_SECURE_ANNOUNCER;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (publications_secure_reader_.first != nullptr) && assign_secure_endpoints) {
        temp_writer_proxy_data->guid().entityId = sedp_builtin_publications_secure_writer;
        temp_writer_proxy_data->set_persistence_entity_id(sedp_builtin_publications_secure_writer);

        if ((!mp_RTPSParticipant->security_manager().discovered_builtin_writer(
                publications_secure_reader_.first->getGuid(), pdata.m_guid, *temp_writer_proxy_data,
                publications_secure_reader_.first->getAttributes().security_attributes()))) {
            elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR,
                      "Security manager returns an error for reader " << publications_secure_reader_.first->getGuid());
        }
    }

    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_PUBLICATION_SECURE_DETECTOR;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (publications_secure_writer_.first != nullptr) && assign_secure_endpoints) {
        temp_reader_proxy_data->guid().entityId = sedp_builtin_publications_secure_reader;
        if ((!mp_RTPSParticipant->security_manager().discovered_builtin_reader(
                publications_secure_writer_.first->getGuid(), pdata.m_guid, *temp_reader_proxy_data,
                publications_secure_writer_.first->getAttributes().security_attributes()))) {
            elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR,
                      "Security manager returns an error for writer " << publications_secure_writer_.first->getGuid());
        }
    }

    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_SECURE_ANNOUNCER;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (subscriptions_secure_reader_.first != nullptr) && assign_secure_endpoints) {
        temp_writer_proxy_data->guid().entityId = sedp_builtin_subscriptions_secure_writer;
        temp_writer_proxy_data->set_persistence_entity_id(sedp_builtin_subscriptions_secure_writer);

        if ((!mp_RTPSParticipant->security_manager().discovered_builtin_writer(
                subscriptions_secure_reader_.first->getGuid(), pdata.m_guid, *temp_writer_proxy_data,
                subscriptions_secure_reader_.first->getAttributes().security_attributes()))) {
            elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR,
                      "Security manager returns an error for reader " << subscriptions_secure_reader_.first->getGuid());
        }
    }

    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_SECURE_DETECTOR;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (subscriptions_secure_writer_.first != nullptr && assign_secure_endpoints)) {
        temp_reader_proxy_data->guid().entityId = sedp_builtin_subscriptions_secure_reader;
        if ((!mp_RTPSParticipant->security_manager().discovered_builtin_reader(
                subscriptions_secure_writer_.first->getGuid(), pdata.m_guid, *temp_reader_proxy_data,
                subscriptions_secure_writer_.first->getAttributes().security_attributes()))) {
            elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR,
                      "Security manager returns an error for writer " << subscriptions_secure_writer_.first->getGuid());
        }
    }
    // }
#else
    static_cast<void>(assign_secure_endpoints);
#endif  // if HAVE_SECURITY
}

void EDPSimple::removeRemoteEndpoints(ParticipantProxyData* pdata) {
    GUID_t tmp_guid;
    tmp_guid.guidPrefix = pdata->m_guid.guidPrefix;

    uint32_t endp = pdata->m_availableBuiltinEndpoints;
    uint32_t auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_PUBLICATION_ANNOUNCER;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (publications_reader_.first != nullptr))  // Exist Pub Writer and i have pub reader
    {
        tmp_guid.entityId = c_EntityId_SEDPPubWriter;
        logDebug(RTPS_EDP, "Remove SEDP Pub writer " << tmp_guid << " to my Pub reader "
                                                     << publications_reader_.first->getGuid());
        (void)publications_reader_.first->matched_writer_remove(tmp_guid);
    }
    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_PUBLICATION_DETECTOR;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (publications_writer_.first != nullptr))  // Exist Pub Detector
    {
        tmp_guid.entityId = c_EntityId_SEDPPubReader;
        logDebug(RTPS_EDP, "Remove SEDP Pub reader " << tmp_guid << " to my Pub writer "
                                                     << publications_writer_.first->getGuid());
        (void)publications_writer_.first->matched_reader_remove(tmp_guid);
    }
    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_ANNOUNCER;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (subscriptions_reader_.first != nullptr))  // Exist Pub Announcer
    {
        tmp_guid.entityId = c_EntityId_SEDPSubWriter;
        logDebug(RTPS_EDP, "Remove SEDP Sub writer " << tmp_guid << " to my Sub reader "
                                                     << subscriptions_reader_.first->getGuid());
        (void)subscriptions_reader_.first->matched_writer_remove(tmp_guid);
    }
    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_DETECTOR;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0) && (subscriptions_writer_.first != nullptr))  // Exist Pub Announcer
    {
        tmp_guid.entityId = c_EntityId_SEDPSubReader;
        logDebug(RTPS_EDP, "Remove SEDP Sub Reader " << tmp_guid << " to my Sub Writer "
                                                     << subscriptions_writer_.first->getGuid());
        (void)subscriptions_writer_.first->matched_reader_remove(tmp_guid);
    }

#if HAVE_SECURITY
    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_PUBLICATION_SECURE_ANNOUNCER;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (publications_secure_reader_.first != nullptr)) {
        tmp_guid.entityId = sedp_builtin_publications_secure_writer;
        logDebug(RTPS_EDP, "Remove SEDP Pub writer " << tmp_guid << " to my Pub reader "
                                                     << publications_secure_reader_.first->getGuid());
        if (publications_secure_reader_.first->matched_writer_remove(tmp_guid)) {
            mp_RTPSParticipant->security_manager().remove_writer(publications_secure_reader_.first->getGuid(),
                                                                 pdata->m_guid, tmp_guid);
        }
    }

    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_PUBLICATION_SECURE_DETECTOR;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (publications_secure_writer_.first != nullptr)) {
        tmp_guid.entityId = sedp_builtin_publications_secure_reader;
        logDebug(RTPS_EDP, "Remove SEDP Pub reader " << tmp_guid << " to my Pub writer "
                                                     << publications_secure_writer_.first->getGuid());
        if (publications_secure_writer_.first->matched_reader_remove(tmp_guid)) {
            mp_RTPSParticipant->security_manager().remove_reader(publications_secure_writer_.first->getGuid(),
                                                                 pdata->m_guid, tmp_guid);
        }
    }

    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_SECURE_ANNOUNCER;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (subscriptions_secure_reader_.first != nullptr)) {
        tmp_guid.entityId = sedp_builtin_subscriptions_secure_writer;
        logDebug(RTPS_EDP, "Remove SEDP Sub Writer " << tmp_guid << " to my Sub Reader "
                                                     << subscriptions_secure_reader_.first->getGuid());
        if (subscriptions_secure_reader_.first->matched_writer_remove(tmp_guid)) {
            mp_RTPSParticipant->security_manager().remove_writer(subscriptions_secure_reader_.first->getGuid(),
                                                                 pdata->m_guid, tmp_guid);
        }
    }
    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_SECURE_DETECTOR;
    // FIXME: FIX TO NOT FAIL WITH BAD BUILTIN ENDPOINT SET
    // auxendp = 1;
    if ((auxendp != 0U) && (subscriptions_secure_writer_.first != nullptr)) {
        tmp_guid.entityId = sedp_builtin_subscriptions_secure_reader;
        logDebug(RTPS_EDP, "Remove SEDP Sub Reader " << tmp_guid << " to my Sub Writer "
                                                     << subscriptions_secure_writer_.first->getGuid());
        if (subscriptions_secure_writer_.first->matched_reader_remove(tmp_guid)) {
            mp_RTPSParticipant->security_manager().remove_reader(subscriptions_secure_writer_.first->getGuid(),
                                                                 pdata->m_guid, tmp_guid);
        }
    }
#endif  // if HAVE_SECURITY
}

bool EDPSimple::areRemoteEndpointsMatched(const ParticipantProxyData* pdata) {
    uint32_t endp = pdata->m_availableBuiltinEndpoints;

    uint32_t auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_PUBLICATION_ANNOUNCER;
    if (auxendp != 0 && publications_reader_.first != nullptr)  // Exist Pub Writer and I have Pub Reader
    {
        GUID_t wguid;
        wguid.guidPrefix = pdata->m_guid.guidPrefix;
        wguid.entityId = c_EntityId_SEDPPubWriter;

        if (!publications_reader_.first->matched_writer_is_matched(wguid)) {
            return false;
        }
    }

    auxendp = endp;  //LCOV_EXCL_START
    auxendp &= DISC_BUILTIN_ENDPOINT_PUBLICATION_DETECTOR;
    if (auxendp != 0 && publications_writer_.first != nullptr)  // Exist Pub Detector
    {
        GUID_t rguid;
        rguid.guidPrefix = pdata->m_guid.guidPrefix;
        rguid.entityId = c_EntityId_SEDPPubReader;

        if (!publications_writer_.first->matched_reader_is_matched(rguid)) {
            return false;
        }
    }

    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_ANNOUNCER;
    if (auxendp != 0 && subscriptions_reader_.first != nullptr)  // Exist Pub Announcer
    {
        GUID_t wguid;
        wguid.guidPrefix = pdata->m_guid.guidPrefix;
        wguid.entityId = c_EntityId_SEDPSubWriter;

        if (!subscriptions_reader_.first->matched_writer_is_matched(wguid)) {
            return false;
        }
    }

    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_DETECTOR;
    if (auxendp != 0 && subscriptions_writer_.first != nullptr)  // Exist Pub Announcer
    {
        GUID_t rguid;
        rguid.guidPrefix = pdata->m_guid.guidPrefix;
        rguid.entityId = c_EntityId_SEDPSubReader;

        if (!subscriptions_writer_.first->matched_reader_is_matched(rguid)) {
            return false;  //LCOV_EXCL_STOP
        }
    }

    return true;
}

#if HAVE_SECURITY
bool EDPSimple::pairing_remote_writer_with_local_builtin_reader_after_security(
    const GUID_t& local_reader, const WriterProxyData& remote_writer_data) {
    bool returned_value = false;

    if (local_reader.entityId == sedp_builtin_publications_secure_reader) {
        (void)publications_secure_reader_.first->matched_writer_add(const_cast<WriterProxyData&>(remote_writer_data));
        returned_value = true;
    } else if (local_reader.entityId == sedp_builtin_subscriptions_secure_reader) {
        (void)subscriptions_secure_reader_.first->matched_writer_add(const_cast<WriterProxyData&>(remote_writer_data));
        returned_value = true;
    }

    return returned_value;
}

bool EDPSimple::pairing_remote_reader_with_local_builtin_writer_after_security(
    const GUID_t& local_writer, const ReaderProxyData& remote_reader_data) {
    bool returned_value = false;

    if (local_writer.entityId == sedp_builtin_publications_secure_writer) {
        (void)publications_secure_writer_.first->matched_reader_add(const_cast<ReaderProxyData&>(remote_reader_data));
        returned_value = true;
    } else if (local_writer.entityId == sedp_builtin_subscriptions_secure_writer) {
        (void)subscriptions_secure_writer_.first->matched_reader_add(const_cast<ReaderProxyData&>(remote_reader_data));
        returned_value = true;
    }

    return returned_value;
}

#endif  // if HAVE_SECURITY

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */
